package com.archstudy.springcloud.websocket.service.impl;

import com.archstudy.push.platform.contract.constants.MessageTypeEnum;
import com.archstudy.springcloud.standard.context.SpringContextUtil;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.client.consumer.DefaultMQPushConsumer;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyContext;
import org.apache.rocketmq.client.consumer.listener.ConsumeConcurrentlyStatus;
import org.apache.rocketmq.client.consumer.listener.MessageListenerConcurrently;
import org.apache.rocketmq.client.exception.MQClientException;
import org.apache.rocketmq.client.producer.DefaultMQProducer;
import org.apache.rocketmq.common.consumer.ConsumeFromWhere;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.common.protocol.heartbeat.MessageModel;
import org.springframework.beans.BeansException;
import org.springframework.beans.factory.annotation.Value;
import org.springframework.context.ApplicationContext;
import org.springframework.context.ApplicationContextAware;
import org.springframework.stereotype.Component;

import javax.annotation.Resource;
import java.util.List;

@Component
@Slf4j
public class PushMessage2NettyChannelService implements ApplicationContextAware {

    private String brokerClusterName = "DefaultCluster";


    @Value("${rocketmq.nameServer}")
    private String rocketmqAddress;
    @Value("${rocketmq.consumer.group}")
    private String consumerGroup;
    @Value("${rocketmq.producer.group}")
    private String producerGroup;

    @Value("${rocketmq.producer.topic}")
    private String topic;


    @Value("${rocketmq.producer.relayTopic}")
    private String relayTopic;

    @Value("${rocketmq.consumer.relayGroup}")
    private String consumerRelayGroup;


    @Value("${rocketmq.producer.concurrent}")
    private int concurrent = 1;

    @Value("${rocketmq.producer.retryTimes}")
    private int retryTimes = 1;

    @Value("${rocketmq.producer.maxWaited}")
    private int maxWaited = 100000;

    private volatile boolean inited = false;


    DefaultMQPushConsumer consumer = null;


    @Resource
    WebserviceMsgProcessImpl webserviceMsgProcess;

    @Override
    public void setApplicationContext(ApplicationContext applicationContext) throws BeansException {
        SpringContextUtil.setContext(applicationContext);
        initRelayConsumerAndProducer();
    }

    private synchronized void initRelayConsumerAndProducer() {
        if (inited) return;
        inited = true;

        startConsumer();
    }

    DefaultMQProducer producer = null;


    private void startConsumer() {


        /**
         * Consumer Group,非常重要的概念，后续会慢慢补充
         */
        consumer = new DefaultMQPushConsumer(consumerGroup);
        consumer.setMessageModel(MessageModel.BROADCASTING);

        //指定NameServer地址，多个地址以 ; 隔开
        consumer.setNamesrvAddr(rocketmqAddress); //修改为自己的
//        consumer.setNamesrvAddr("192.168.116.115:9876;192.168.116.116:9876"); //修改为自己的


/**
 * 设置Consumer第一次启动是从队列头部开始消费还是队列尾部开始消费
 * 如果非第一次启动，那么按照上次消费的位置继续消费
 */
        consumer.setConsumeFromWhere(ConsumeFromWhere.CONSUME_FROM_FIRST_OFFSET);

        try {
            String topicKey = MessageTypeEnum.WEB_SOCKET.getTopicKey(topic);
            consumer.subscribe(topicKey, MessageTypeEnum.WEB_SOCKET.name() /*"websocket"*/);
        } catch (MQClientException e) {
            e.printStackTrace();
            return;
        }

        consumer.registerMessageListener(new MessageListenerConcurrently() {
            @Override
            public ConsumeConcurrentlyStatus consumeMessage(List<MessageExt> msgs, ConsumeConcurrentlyContext context) {

                try {
                    for (int i = 0; i < msgs.size(); i++) {
                        MessageExt msg = msgs.get(i);
//                    String content = new String(msg.getBody());
//                    log.info("收到消息：{}", msg.getMsgId() + " " + msg.getTopic() + " " + msg.getTags() + " " + content);
                        String json = new String(msg.getBody(), "UTF-8");
                        log.debug("json = " + json);
                        webserviceMsgProcess.updateAndSend(json);
                    }
                    return ConsumeConcurrentlyStatus.CONSUME_SUCCESS;

                } catch (Throwable throwable) {
                    throwable.printStackTrace();
                    return ConsumeConcurrentlyStatus.RECONSUME_LATER;

                }

            }
        });
        try {
            consumer.start();
        } catch (MQClientException e) {
            e.printStackTrace();
        }

    }


}
